package com.nian.hbase.base.service.impl;

import com.nian.hbase.base.aop.ParsingAnnotation;
import com.nian.hbase.base.model.HBaseRowModel;
import com.nian.hbase.base.model.HBaseTableModel;
import com.nian.hbase.base.service.IHBaseService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;

/**
 * @author zhangmj
 * @date 2020/4/19 14:33
 */
@Slf4j
@Service
public class HBaseServiceImpl<T> implements IHBaseService<T> {

    @Autowired
    Connection hBaseConnect;

    @Override
    public T put(T model) {
        Table table = null;
        HBaseTableModel tableModel = null;
        try {
            tableModel = ParsingAnnotation.readModel(model);
            table = hBaseConnect.getTable(TableName.valueOf(tableModel.getTableName()));
            this.put(tableModel, table);
            return model;
        } catch (TableNotFoundException e) {
            log.error("写入数据时, 未找到表, 做新建表操作，空间为默认空间", e);
            if(tableModel.getFlag() == 1){
                // 1 创建表, 默认表空间
                this.createTable(tableModel.getTableName(), tableModel.getFamilyList());
                // 2
                try {
                    this.put(tableModel, table);
                } catch (IOException ex) {
                    log.error("二次写入数据错误：", ex);
                }
            }
        } catch (IOException e) {
            log.error("保存信息错误:", e);
        }finally {
            close(table);
        }
        return null;
    }

    public void put(HBaseTableModel tableModel, Table table) throws IOException {
        tableModel.addFlag();
        List<HBaseRowModel> columnList = tableModel.getColumnList();
        for (HBaseRowModel row:columnList) {
            Put put = new Put(Bytes.toBytes(row.getRowName()));
            put.addColumn(Bytes.toBytes(row.getFamily()),Bytes.toBytes(row.getQualifier()), Bytes.toBytes(row.getValue()));
            table.put(put);
        }
    }
    /**
     * 写入一行、一列的值
     * @param tableName
     * @param rowName
     * @param familyName
     * @param qualifier
     * @param value
     */
    @Override
    public void put(String tableName, String rowName, String familyName, String qualifier, String value) {
        Table table = null;
        try {
            table = hBaseConnect.getTable(TableName.valueOf(tableName));
            Put put = new Put(Bytes.toBytes(rowName));
            put.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(qualifier), Bytes.toBytes(value));
            table.put(put);
        } catch (IOException e) {
            log.error("写入一行、一列的值异常:", e);
        } finally {
            close(table);
        }
    }
    /**
     * 获取某一行、某一列的值
     * @param tableName
     * @param rowName
     * @param family
     * @param qualifier
     * @return
     */
    @Override
    public String get(String tableName, String rowName, String family, String qualifier) {
        Table table = null;
        try {
            table = hBaseConnect.getTable(TableName.valueOf(tableName));
            Get get = new Get(Bytes.toBytes(rowName));
            Result rs = table.get(get);
            Cell[] cells = rs.rawCells();
            for (Cell c:cells) {
                String familyN = Bytes.toString(c.getFamilyArray(), c.getFamilyOffset(), c.getFamilyLength());
                String qualifierN = Bytes.toString(c.getQualifierArray(), c.getQualifierOffset(), c.getQualifierLength());
                if(StringUtils.equals(family, familyN) && StringUtils.equals(qualifier, qualifierN)){
                    return Bytes.toString(c.getValueArray(), c.getValueOffset(), c.getValueLength());
                }
            }
        } catch (IOException e) {
            log.error("获取某一行、某一列的值异常:", e);
        }finally {
            close(table);
        }
        return null;
    }

    /**
     * 扫描表
     * @param tableName
     * @return
     */
    @Override
    public ResultScanner scanner(String tableName) {
        Table table = null;
        try {
            table = hBaseConnect.getTable(TableName.valueOf(tableName));
            Scan scan = new Scan();
            scan.setCaching(1000);
            return table.getScanner(scan);
        } catch (IOException e) {
            log.error("扫描表异常:", e);
        }finally {
            close(table);
        }
        return null;
    }

    @Override
    public List<T> scanner(Class<T> clazz) {
        ResultScanner scanner = null;
        try {
            HBaseTableModel model = ParsingAnnotation.readModel(clazz);
            if(model != null){
                String tableName = model.getTableName();
                scanner = scanner(tableName);
                Result rs = null;
                List newList = new LinkedList<>();
                while ((rs = scanner.next()) != null) {
                    List<Cell> cells = rs.listCells();
                    List<HBaseRowModel> list = new LinkedList<>();
                    for (Cell cell : cells) {
                        HBaseRowModel rowModel = new HBaseRowModel(tableName, Bytes.toString(CellUtil.cloneRow(cell)),
                                Bytes.toString(CellUtil.cloneFamily(cell)),
                                Bytes.toString(CellUtil.cloneQualifier(cell)),
                                Bytes.toString(CellUtil.cloneValue(cell)) );
                        list.add(rowModel);
                    }
                    T newModel = ParsingAnnotation.initModel(clazz, list);
                    newList.add(newModel);
                }
                return newList;
            }
        }catch (Exception e){
            log.error("扫描表 Class 异常:", e);
        }finally {
            close(scanner);
        }

        return null;
    }

    /**
     * 利用过滤器获取数据
     * @param tableName
     * @param rowName
     * @param filterList
     * @return
     */
    @Override
    public Result getRow(String tableName, String rowName, FilterList filterList) {
        Table table = null;
        try {
            table = hBaseConnect.getTable(TableName.valueOf(tableName));
            Get get = new Get(Bytes.toBytes(rowName));
            get.setFilter(filterList);
            return table.get(get);
        } catch (IOException e) {
            log.error("利用过滤器获取数据 异常:", e);
        }finally {
            close(table);
        }
        return null;
    }
    /**
     * 删除行数据
     * @param tableName
     * @param rowName
     * @return
     */
    @Override
    public boolean deleteRow(String tableName, String rowName) {
        Table table = null;
        try {
            table = hBaseConnect.getTable(TableName.valueOf(tableName));
            Delete delete = new Delete(Bytes.toBytes(rowName));
            table.delete(delete);
            return true;
        } catch (IOException e) {
            log.error("删除行数据 异常:", e);
        }finally {
            close(table);
        }
        return false;
    }
    /**
     * 删除列族
     * @param tableName
     * @param familyName
     * @return
     */
    @Override
    public boolean deleteColumnFamily(String tableName, String familyName) {
        Admin admin = null;
        try {
            admin = hBaseConnect.getAdmin();
            admin.deleteColumnFamily(TableName.valueOf(tableName), Bytes.toBytes(familyName));
            return true;
        } catch (IOException e) {
            log.error("删除列族 异常:", e);
        }finally {
            close(admin);
        }
        return false;
    }
    /**
     * 删除列
     * @param tableName
     * @param rowName
     * @param familyName
     * @param qualifier
     * @return
     */
    @Override
    public boolean deleteQualifier(String tableName, String rowName, String familyName, String qualifier) {
        Table table = null;
        try {
            table = hBaseConnect.getTable(TableName.valueOf(tableName));
            Delete delete = new Delete(Bytes.toBytes(rowName));
            delete.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(qualifier));
            table.delete(delete);
            return true;
        } catch (IOException e) {
            log.error("删除列 异常:", e);
        }finally {
            close(table);
        }
        return false;
    }
    /**
     * 创建表名，若有表空间
     * @param nameSpace
     * @param tableName
     * @param columns
     * @return
     */
    @Override
    public boolean createTable(String nameSpace, String tableName, String... columns) {
        Admin admin = null;
        try {
            admin = hBaseConnect.getAdmin();
            NamespaceDescriptor namespaceDescriptor = admin.getNamespaceDescriptor(nameSpace);
            if(namespaceDescriptor == null){
                namespaceDescriptor = NamespaceDescriptor.create(nameSpace).build();
            }
            TableDescriptorBuilder mk = TableDescriptorBuilder.newBuilder(TableName.valueOf(nameSpace +":"+tableName));
            List<ColumnFamilyDescriptor> columnFamilyDescriptors = buildColumnFamily(columns);
            mk.setColumnFamilies(columnFamilyDescriptors);
            admin.createTable(mk.build());
            return true;
        } catch (IOException e) {
            log.error("创建表 异常:", e);
        }finally {
            close(admin);
        }
        return false;
    }

    @Override
    public boolean createTable(String tableName, String... columns) {
        Admin admin = null;
        try {
            admin = hBaseConnect.getAdmin();
            TableDescriptorBuilder mk = TableDescriptorBuilder.newBuilder(TableName.valueOf(tableName));
            List<ColumnFamilyDescriptor> columnFamilyDescriptors = buildColumnFamily(columns);
            mk.setColumnFamilies(columnFamilyDescriptors);
            admin.createTable(mk.build());
            return true;
        } catch (IOException e) {
            log.error("创建表异常：", e);
        }finally {
            close(admin);
        }
        return false;
    }

    @Override
    public boolean createNameSpace(String nameSpace) {
        Admin admin = null;
        try {
            NamespaceDescriptor.create(nameSpace).build();
            admin = hBaseConnect.getAdmin();
            return true;
        } catch (IOException e) {
            log.error("创建表空间 异常:", e);
        }finally {
            close(admin);
        }
        return false;
    }

    public List<ColumnFamilyDescriptor> buildColumnFamily(String[] columns) {
        List<ColumnFamilyDescriptor> list = new ArrayList<>(columns.length);
        for (String c:columns) {
            ColumnFamilyDescriptorBuilder column = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(c));
            list.add(column.build());
        }
        return list;
    }

    @Override
    public void close(Admin admin) {
        try {
            if(admin != null){
                admin.close();
            }
        } catch (IOException e) {
            log.error("关闭 Admin 异常:", e);
        }
    }

    @Override
    public void close(Table table) {
        try {
            if(table != null){
                table.close();
            }
        } catch (IOException e) {
            log.error("关闭 table 异常:", e);
        }
    }

    @Override
    public void close(Admin admin, Table table) {
        this.close(table);
        this.close(admin);
    }

    @Override
    public void close(ResultScanner scanner) {
        try {
            if(scanner != null){
                scanner.close();
            }
        } catch (Exception e) {
            log.error("关闭 ResultScanner 异常:", e);
        }
    }
}
